package com.baishancloud.log.quality

import cn.hutool.core.exceptions.ExceptionUtil
import cn.hutool.json.{JSONObject, JSONUtil}
import com.baishancloud.log.common.entity._
import com.baishancloud.log.common.util.{ExternalDataFetchUtil, StringUtil, TimeUtil}
import com.baishancloud.log.format.java.util.IPLocator
import com.baishancloud.log.quality.pojo.{MobileLog, StarRocksMobile, Tags}
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.slf4j.{Logger, LoggerFactory}

import java.lang
import java.util.{Timer, TimerTask}
import scala.collection.mutable

/**
 * 对输入的移动端日志进行对象转换<br>
 * 输出为为二元组(MobileLog, StarRocksMobile)，第一个元素为计算比率的MobileLog，第二个值为写入doris明细数据的MobileLog对象
 *
 * @author ziqiang.wang 
 * @date 2021/11/19 17:55
 */
class MobileLogParse(business: String) extends RichMapFunction[String, (MobileLog, StarRocksMobile)] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[MobileLogParse])

  //服务端cache组信息表
  private val serverCacheGroupInfo: mutable.Map[String, ServerCacheGroupEntity] = mutable.Map[String, ServerCacheGroupEntity]()
  // 域名id信息表
  private val domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
  // 国家id信息表
  private val countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
  // 省份id信息表
  private val provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
  // 运营商id信息表
  private val ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()

  private var subTaskIndex: Int = -1
  //允许迟到数据最长时间
  private var allowOldest: Long = System.currentTimeMillis() - 24 * 3600 * 1000

  /**
   * 初始化缓存数据，增加定时器，定时更新缓存数据
   */
  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    ExternalDataFetchUtil.serverCacheGroupFetch(serverCacheGroupInfo)
    ExternalDataFetchUtil.domainInfoFetch(domainInfo)
    ExternalDataFetchUtil.countryInfoFetch(countryInfo)
    ExternalDataFetchUtil.provinceInfoFetch(provinceInfo)
    ExternalDataFetchUtil.ispInfoFetch(ispInfo)
    LOGGER.info(s"serverCacheGroupInfo${subTaskIndex}初始化完成，ipCache表size为：${serverCacheGroupInfo.size}")
    LOGGER.info(s"domainInfo${subTaskIndex}初始化完成，domainId表size为：${domainInfo.size}")
    LOGGER.info(s"countryInfo${subTaskIndex}初始化完成，countryId表size为：${countryInfo.size}")
    LOGGER.info(s"provinceInfo${subTaskIndex}初始化完成，provinceId表size为：${provinceInfo.size}")
    LOGGER.info(s"ispInfo${subTaskIndex}初始化完成，ispId为：${ispInfo.size}")
    val timer: Timer = new Timer("缓存数据更新", true)
    //增加定时器，每半小时获取一次最新缓存数据
    timer.scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = {
        ExternalDataFetchUtil.serverCacheGroupFetch(serverCacheGroupInfo)
        ExternalDataFetchUtil.domainInfoFetch(domainInfo)
        ExternalDataFetchUtil.countryInfoFetch(countryInfo)
        ExternalDataFetchUtil.provinceInfoFetch(provinceInfo)
        ExternalDataFetchUtil.ispInfoFetch(ispInfo)
        allowOldest = System.currentTimeMillis() - 24 * 3600 * 1000
      }
    }, 1800 * 1000, 1800 * 1000)
  }

  override def map(value: String): (MobileLog, StarRocksMobile) = {
    try {
      val jsonObj: JSONObject = JSONUtil.parseObj(value)
      val timeIso: lang.Long = jsonObj.getLong("time_iso")
      //过滤迟到数据
      if (timeIso < allowOldest) return (null, null)
      val extendedFields: JSONObject = jsonObj.getJSONObject("extended_fields")
      val split: Array[String] = extendedFields.getStr("video_url").split("/")
      val domain: String = StringUtil.subString(if (split.length > 2) split(2) else "", 200)
      //ip信息
      val ipInfo = IPLocator.lookup(jsonObj.getStr("ip"))
      var isp: String = ipInfo.getIsp.split("/")(0)
      if (isp.equals("other")) isp = "其他"
      if (isp.contains("广电")) {
        isp = "广电"
      } else if (isp.contains("华数")) {
        isp = "华数"
      }
      val country: String = StringUtil.subString(if (ipInfo.getCountry.equals("other")) "其他" else ipInfo.getCountry, 33)
      var province: String = StringUtil.subString(if (ipInfo.getProvince.equals("other")) "其他" else ipInfo.getProvince, 33)
      if ((country.equals("中国") || country.startsWith("CHINA")) && province.equals("其他")) province = "中国其他地区"

      val cacheGroup: String = StringUtil.subString(serverCacheGroupInfo.getOrElse(extendedFields.getStr("video_ip"), ServerCacheGroupEntity.DEFAULT).cacheGroupCode, 200)
      (
        MobileLog(
          Tags(
            duration,
            platformJudge(jsonObj.getInt("platform")),
            business,
            sourceMobile,
            domain,
            province,
            isp,
            netTypeJudge(jsonObj.getInt("network")),
            cacheGroup,
            ""
          ),
          timeIso,
          extendedFields.getInt("first_video_time"),
          extendedFields.getDouble("asset_item_time_of_session"),
          extendedFields.getDouble("buffering_count"),
          extendedFields.getDouble("time_of_video"),
          extendedFields.getStr("asset_item_session")
        ),
        StarRocksMobile(
          business,
          TimeUtil.timeAtLong(timeIso, 60 * 1000) / 1000,
          TimeUtil.timeAtString(timeIso, 60 * 1000),
          domainInfo.getOrElse(domain, DomainEntity.DEFAULT).domainId,
          domain,
          platformJudge(jsonObj.getInt("platform")),
          serverCacheGroupInfo.getOrElse(extendedFields.getStr("video_ip"), ServerCacheGroupEntity.DEFAULT).hostId,
          serverCacheGroupInfo.getOrElse(extendedFields.getStr("video_ip"), ServerCacheGroupEntity.DEFAULT).cacheGroupCode,
          netTypeJudge(jsonObj.getInt("network")),
          extendedFields.getShort("video_http_code"),
          ispInfo.getOrElse(isp, IspEntity.DEFAULT).id,
          isp,
          countryInfo.getOrElse(country, CountryEntity.DEFAULT).id,
          country,
          provinceInfo.getOrElse(province, ProvinceEntity.DEFAULT).id,
          province,
          extendedFields.getStr("asset_item_session"),
          extendedFields.getInt("asset_item_time_of_session"),
          extendedFields.getInt("first_video_time"),
          extendedFields.getInt("time_of_video"),
          extendedFields.getInt("buffering_count")
        )
      )
    } catch {
      case e: Exception => {
        LOGGER.error(s"移动${business}-json内容提取出错，数据为：\n${value}\n错误信息为：" + ExceptionUtil.stacktraceToString(e, 1000))
        (null, null)
      }
    }
  }

  /**
   * 判断网络类型
   *
   * @param network 网络类型id
   * @return 网络类型
   */
  private def netTypeJudge(network: Int): String = {
    network match {
      case 0 => "UNKNOWN"
      case 1 => "WIFI"
      case 2 => "CELLULAR"
      case 3 => "OFFLINE"
      case 4 => "OTHERNET"
      case 5 => "ETHERNET"
      case _ => ""
    }
  }

  /**
   * 判断移动平台
   *
   * @param platform 平台id
   * @return 移动平台类型
   */
  private def platformJudge(platform: Int): String = {
    platform match {
      case 1 => "iphone"
      case 2 => "ipad"
      case 3 => "android"
      case _ => ""
    }
  }
}
